package com.feiwei

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.types.Row
object day10_TableEventTime {


  def main(args: Array[String]): Unit = {


    val environment = StreamExecutionEnvironment.getExecutionEnvironment
      environment.setParallelism(1)
      environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode().build()
    val path="E:\\repository\\company\\myself\\flink-learning\\flink-learning-demo\\src\\main\\resources\\sensor.txt"

    val tableEnv= StreamTableEnvironment.create(environment,settings)

    val readStream = environment.readTextFile(path)

    val sensor=  readStream.map(v=>{

      val arr = v.split(",")

       SensorReading(arr(0),arr(1).toLong,arr(2).toDouble)
    })
      //设置时间和水位线,设置为乱序，延迟时间为1秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp*1000
      })


    //如果流中指定了水位线，在转换成表的时候，直接就可以使用字段.rowtime
  var tq=  tableEnv.fromDataStream(sensor,'id,'temperature,'timestamp.rowtime)

    tq.toAppendStream[Row].print()
    environment.execute()

  }
}
